#pragma once
#include "../common/detail.hpp"
#include "../common/net.hpp"
#include <memory>
#include <mutex>
#include <unordered_set>
#include <vector>


namespace rpc
{
    namespace server
    {
        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;

            void onTopicRequest(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                //主题的创建
                //主题的删除
                //主题的订阅
                //主题的取消订阅
                //主题消息的发布
                TopicOptype topic_optype = msg->optype();
                bool ret = true;
                switch(topic_optype)
                {
                    case TopicOptype::TOPIC_CREATE : topicCreate(msg); break;
                    case TopicOptype::TOPIC_REMOVE : topicRemove(msg); break;
                    case TopicOptype::TOPIC_SUBSCRIBE : ret = topicSubscribe(conn, msg); break;
                    case TopicOptype::TOPIC_PUBLISH : ret = topicPush(conn, msg); break;
                    case TopicOptype::TOPIC_CANCEL : topicCancel(conn, msg); break;
                    default : return errorResponse(conn, msg, RCode::RCODE_INVALID_OPTYPE); 
                }
                if(!ret)
                {
                    return errorResponse(conn, msg, RCode::RCODE_NOT_FOUND_TOPIC);
                }
                return topicResponse(conn, msg);
            }
            //订阅者断开时的操作
            void onShutdown(const BaseConnection::ptr &conn)
            {
                //消息发布者断开连接，不需要进行任何操作； 消息订阅者断开连接需要删除管理数据
                std::vector<Topic::ptr> topics;
                Subscriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _subscribers.find(conn);
                    if(it == _subscribers.end())
                    {
                        ILOG("断开的连接不是消息订阅者");
                        return;
                    }
                    subscriber = it->second;
                    //获取到订阅者退出后，受影响的主题对象
                    for(auto &topic_name : subscriber->topics)
                    {
                        auto topic_it = _topics.find(topic_name);
                        if(topic_it == _topics.end())
                        {
                            ELOG("从订阅对象中找不到订阅者订阅的主题");
                            continue;
                        }
                        topics.emplace_back(topic_it->second);
                    }
                    //从订阅者映射信息中，删除映射关系
                    _subscribers.erase(it);
                }
                //从受影响的主题对象中，删除主题与订阅者的映射关系
                for(auto &topic : topics)
                {
                    topic->removeSubscriber(subscriber);
                }
            }
        private:
            void errorResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg, RCode rcode)
            {
                auto msg_rsp = MessageFactory::create<TopicResponse>();
                msg_rsp->setId(msg->rid());
                msg_rsp->setMtype(MType::RSP_TOPIC);
                msg_rsp->setRCode(rcode);
                return conn->send(msg_rsp);
            }
            void topicResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                auto msg_rsp = MessageFactory::create<TopicResponse>();
                msg_rsp->setId(msg->rid());
                msg_rsp->setMtype(MType::RSP_TOPIC);
                msg_rsp->setRCode(RCode::RCODE_OK);
                return conn->send(msg_rsp);
            }
            void topicCreate(const TopicRequest::ptr &msg)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                std::string topic_name = msg->topicKey();
                auto topic = std::make_shared<Topic>(topic_name);
                _topics.insert({topic_name, topic});
            }   
            void topicRemove(const TopicRequest::ptr &msg)
            {
                //1.查看当前主题，查找有哪些订阅者，从订阅者中将主题信息删除
                //2.删除主题名称与主题对象的映射关系
                Topic::ptr topic;
                std::string topic_name = msg->topicKey();
                std::unordered_set<Subscriber::ptr> subscribers;

                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _topics.find(topic_name);
                    if(it == _topics.end())
                    {
                        ELOG("找不到待删除主题对象 : %s", topic_name.c_str());
                        return;
                    }
                    topic = it->second;
                    subscribers = topic->subscribers;
                    //操作2
                    _topics.erase(it);
                    for(auto &subscriber : subscribers)
                    {
                        //操作1
                        subscriber->removeTopic(topic_name);
                    }

                }

            }  
            bool topicSubscribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                //先找出主题对象，以及订阅者对象
                //如果没找到主题，则报错；如果没找到订阅者，则创建订阅者对象
                Topic::ptr topic;
                Subscriber::ptr subscriber; 

                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(msg->topicKey());
                    if(topic_it == _topics.end())
                    {
                        DLOG("找不到主题！");
                        return false;
                    }
                    topic = topic_it->second;
                    auto sub_it = _subscribers.find(conn);
                    if(sub_it != _subscribers.end())
                    {
                        subscriber = sub_it->second;
                    }
                    else
                    {
                        subscriber = std::make_shared<Subscriber>(conn);
                        _subscribers.insert({conn, subscriber});
                    }
                }
                //在主题对象中新增一个订阅对象连接；在订阅者中新增一个主题订阅
                topic->appendSubscriber(subscriber);
                subscriber->appendTopic(topic->topic_name);
                return true;
            }
            void topicCancel(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                Topic::ptr topic;
                Subscriber::ptr subscriber;

                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(msg->topicKey());
                    if(topic_it != _topics.end())
                    {
                        topic = topic_it->second;
                    }
                    auto sub_it = _subscribers.find(conn);
                    if(sub_it != _subscribers.end())
                    {
                        subscriber = sub_it->second;
                    }
                    //从主题对象中删除订阅者链接， 从订阅者对象中删除主题
                    if(subscriber){subscriber->removeTopic(msg->topicKey());}
                    if(topic && subscriber){topic->removeSubscriber(subscriber);}
                }
            }
            bool topicPush(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                Topic::ptr topic;

                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(msg->topicKey());
                    if(topic_it == _topics.end())
                    {
                        return false;
                    }
                    topic = topic_it->second;
                }
                topic->pushMessage(msg);
                return true;
            }

        private:
            struct Subscriber
            {
                using ptr = std::shared_ptr<Subscriber>;
                std::mutex _mutex;
                BaseConnection::ptr conn;
                std::unordered_set<std::string> topics; //该订阅者订阅的主题

                Subscriber(const BaseConnection::ptr &c):conn(c)
                {}
                //订阅时调用
                void appendTopic(const std::string &topic_name)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    topics.insert(topic_name);
                }
                //主题被删除或者取消订阅时候调用
                void removeTopic(const std::string &topic_name)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    topics.erase(topic_name);
                }
            };
            struct Topic
            {
                using ptr = std::shared_ptr<Topic>;
                std::mutex _mutex;
                std::string topic_name; 
                std::unordered_set<Subscriber::ptr> subscribers; //当前主题的订阅者

                Topic(const std::string &name):topic_name(name)
                {}
                //新增订阅时候调用
                void appendSubscriber(const Subscriber::ptr &subscriber)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    subscribers.insert(subscriber);
                }
                //取消订阅或者订阅者下线时候调用
                void removeSubscriber(const Subscriber::ptr &subscriber)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    subscribers.erase(subscriber);
                }
                //收到消息发布请求时候调用
                void pushMessage(const BaseMessage::ptr &msg)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    for(auto subscriber : subscribers)
                    {
                        subscriber->conn->send(msg);
                    }
                }
            };
            
        private:
            std::mutex _mutex;
            std::unordered_map<std::string, Topic::ptr> _topics;
            std::unordered_map<BaseConnection::ptr, Subscriber::ptr> _subscribers;
        };
    }
}